package com.zc.redisson.stream.config;

import com.alibaba.fastjson.JSON;
import com.zc.domain.ElectricAlarm;
import com.zc.redisson.stream.service.ElectricAlarmService;
import com.zc.redisson.stream.service.EnergyAlarmService;
import com.zc.redisson.stream.service.PhoenixAlarmService;
import jodd.util.ThreadUtil;
import org.redisson.api.*;
import org.redisson.api.stream.StreamAddArgs;
import org.redisson.api.stream.StreamReadGroupArgs;
import org.redisson.api.stream.StreamTrimArgs;
import org.redisson.codec.JsonJacksonCodec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import javax.annotation.PreDestroy;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * redisson 接入redis stream配置（redis5.0以上）
 */
//@Configuration
@SuppressWarnings({"rawtypes", "unchecked"})
@ConditionalOnWebApplication
public class RedissonStreamConfig  {

    private final Logger log = LoggerFactory.getLogger(RedissonStreamConfig.class);

    @Autowired
    private RedisTemplate redisTemplate;

    @Value("${redisStream.phoenixAlarmKey}")
    private String phoenixAlarmKey;

    //指标阈值
    @Value("${redisStream.electricAlarmKey}")
    private String electricAlarmKey;

    //用能阈值
    @Value("${redisStream.energyAlarmKey}")
    private String energyAlarmKey;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private PhoenixAlarmService phoenixAlarmService;

    @Autowired
    private ElectricAlarmService electricAlarmService;

    @Autowired
    private EnergyAlarmService energyAlarmService;

    @Autowired
    private final Executor emsThreadPoolTaskExecutor;

    /**
     * 消费者空闲毫秒数,达到后被删除
     */
    private static final Long removeConsumerIdleTime  =  Duration.of(1, ChronoUnit.HOURS).toMillis();


    public RedissonStreamConfig(@Qualifier("emsThreadPoolTaskExecutor") Executor emsThreadPoolTaskExecutor) {
        this.emsThreadPoolTaskExecutor = emsThreadPoolTaskExecutor;
    }


    private boolean isRunning = true;

    @PreDestroy
    public void preDestroy() {
        this.isRunning = false;
    }

    /**
     * TODO-ZC 1. 未确认消息待处理 2. 空闲消费者待处理 3. 项目关闭redisson异常待处理
     */
    @Bean
    public void electricAlarmStream() {
        String groupName = electricAlarmKey + "_group";
        String consumerName = electricAlarmKey + "_consumer" + System.currentTimeMillis();
        RStream<String, ElectricAlarm> electricAlarmStream = redissonClient.getStream(electricAlarmKey,JsonJacksonCodec.INSTANCE);
        if (!electricAlarmStream.isExists()) {
            StreamMessageId streamMessageId = electricAlarmStream.add(StreamAddArgs.entry("0", null));
            electricAlarmStream.remove(streamMessageId);
        }
        //创建一个group，一个group需要在stream数据添加前创建，否则这个group只能读它创建以后写入stream的数据
        List<StreamGroup> list = electricAlarmStream.listGroups();
        long groupcount = list.stream().filter(streamGroup -> streamGroup.getName().equals(groupName)).count();
        if (groupcount == 0) {
            electricAlarmStream.createGroup(groupName);
        }
//        List<PendingEntry> pendingEntries = electricAlarmStream.listPending(groupName, StreamMessageId.MIN, StreamMessageId.MAX, Integer.MAX_VALUE);
//        pendingEntries.forEach(pendingEntry -> {
//            Map<StreamMessageId, Map<String, ElectricMeter>> claimMap = electricAlarmStream.claim(groupName, consumerName, 10, TimeUnit.MINUTES, pendingEntry.getId());
//            log.info("claim消息:{}",claimMap);
//            for (StreamMessageId messageId : claimMap.keySet()) {
//                log.info("messageId:{}", messageId);
//                Map<String, ElectricMeter> messageMap = claimMap.get(messageId);
//                log.info("message:{}", claimMap.get(messageId));
//
//                emsThreadPoolTaskExecutor.execute(new Runnable() {
//                    @Override
//                    public void run() {
//                        electricAlarmService.onMessage(messageMap);
//                    }
//                });
//
//                electricAlarmStream.ack(groupName, messageId);
//                electricAlarmStream.remove(messageId);
//            }
//        });
        List<StreamConsumer> streamConsumers = electricAlarmStream.listConsumers(groupName);
        // 删除空闲消费者
        streamConsumers.forEach(streamConsumer -> {
            if (streamConsumer.getIdleTime() > removeConsumerIdleTime && streamConsumer.getPending() == 0) {
                log.info("删除空闲消费者:{}",streamConsumer.getName());
                electricAlarmStream.removeConsumer(groupName, streamConsumer.getName());
            }
        });
        Thread thread = new Thread(() -> {
            while (isRunning) {
                try {
                    // 获取并移除此队列的头部，在元素变得可用之前一直等待 。queue的长度 == 0 的时候，阻塞等待
                    Map<StreamMessageId, Map<String, ElectricAlarm>> messageIdMapMap = electricAlarmStream.readGroup(groupName, consumerName,
                            StreamReadGroupArgs.neverDelivered().timeout(Duration.of(30, ChronoUnit.SECONDS)).count(20));
                    // 自动认领未确认的消息
                    AutoClaimResult<String, ElectricAlarm> autoClaim = electricAlarmStream.autoClaim(groupName, consumerName, 10, TimeUnit.MINUTES, StreamMessageId.MIN, 20);
//                    log.info("redisson-stream consume message = {}", messageIdMapMap);
                    log.info("消费者数量:{}", electricAlarmStream.listConsumers(groupName).size());
                    log.info("消费者信息:{}", JSON.toJSONString(electricAlarmStream.listConsumers(groupName)));
                    log.info("pendingInfo:{}", JSON.toJSONString(electricAlarmStream.getPendingInfo(groupName)));
                    log.info("listPending:{}", JSON.toJSONString(electricAlarmStream.listPending(groupName, StreamMessageId.MIN, StreamMessageId.MAX, 100)));
                    if (messageIdMapMap == null) {
                        messageIdMapMap = new HashMap<>();
                    }
                    messageIdMapMap.putAll(autoClaim.getMessages());
                    for (StreamMessageId messageId : messageIdMapMap.keySet()) {
                        log.info("messageId:{}", messageId);
                        Map<String, ElectricAlarm> messageMap = messageIdMapMap.get(messageId);
                        log.info("message:{}", messageIdMapMap.get(messageId));

                        emsThreadPoolTaskExecutor.execute(() -> electricAlarmService.onMessage(messageMap));

                        electricAlarmStream.ack(groupName, messageId);
                        electricAlarmStream.remove(messageId);
                    }
                    // 裁剪消息数量
                    electricAlarmStream.trim(StreamTrimArgs.maxLen(1000).noLimit());
                } catch (Exception e) {
                    log.error("electricAlarmStream异常", e);
                    ThreadUtil.sleep(1000);
                }
            }
        });
        thread.setDaemon(true);
        thread.start();
    }


    @Bean
    public void energyAlarmStream() {
        String groupName = energyAlarmKey + "_group";
        String consumerName = energyAlarmKey + "_consumer" + System.currentTimeMillis();
        RStream energyAlarmStream = redissonClient.getStream(energyAlarmKey, JsonJacksonCodec.INSTANCE);
        if (!energyAlarmStream.isExists()) {
            StreamMessageId streamMessageId = energyAlarmStream.add(StreamAddArgs.entry(0, 0));
            energyAlarmStream.remove(streamMessageId);
        }
        //创建一个group，一个group需要在stream数据添加前创建，否则这个group只能读它创建以后写入stream的数据
        List<StreamGroup> list = energyAlarmStream.listGroups();
        long groupcount = list.stream().filter(streamGroup -> streamGroup.getName().equals(groupName)).count();
        if (groupcount == 0) {
            energyAlarmStream.createGroup(groupName);
        }
        Thread thread = new Thread(() -> {
            while (isRunning) {
                try {
                    // 获取并移除此队列的头部，在元素变得可用之前一直等待 。queue的长度 == 0 的时候，阻塞等待
                    Map<StreamMessageId, Map<String, String>> messageIdMapMap = energyAlarmStream.readGroup(groupName, consumerName,
                            StreamReadGroupArgs.neverDelivered().timeout(Duration.of(30, ChronoUnit.SECONDS)).count(20));
                    PendingResult pendingInfo = energyAlarmStream.getPendingInfo(groupName);
//                    log.info("redisson-stream consume message = {}", messageIdMapMap);
                    log.info("消费者数量:{}", energyAlarmStream.listConsumers(groupName).size());
                    if (messageIdMapMap == null) {
                        continue;
                    }
                    for (StreamMessageId messageId : messageIdMapMap.keySet()) {
                        log.info("messageId:{}", messageId);
                        Map<String, String> messageMap = messageIdMapMap.get(messageId);
                        log.info("message:{}", messageIdMapMap.get(messageId));

                        emsThreadPoolTaskExecutor.execute(new Runnable() {
                            @Override
                            public void run() {
                                energyAlarmService.onMessage(messageMap);
                            }
                        });

                        energyAlarmStream.ack(groupName, messageId);
                        energyAlarmStream.remove(messageId);
                    }
                } catch (Exception e) {
                    log.error("energyAlarmStream异常", e);
                    ThreadUtil.sleep(1000);
                }
            }
        });
        thread.setDaemon(true);
        thread.start();
    }

    @Bean
    public void phoenixAlarmStream() {
        String groupName = phoenixAlarmKey + "_group";
        String consumerName = phoenixAlarmKey + "_consumer" + System.currentTimeMillis();
        RStream phoenixAlarmStream = redissonClient.getStream(phoenixAlarmKey);
        if (!phoenixAlarmStream.isExists()) {
            StreamMessageId streamMessageId = phoenixAlarmStream.add(StreamAddArgs.entry(0, 0));
            phoenixAlarmStream.remove(streamMessageId);
        }
        //创建一个group，一个group需要在stream数据添加前创建，否则这个group只能读它创建以后写入stream的数据
        List<StreamGroup> list = phoenixAlarmStream.listGroups();
        long groupcount = list.stream().filter(streamGroup -> streamGroup.getName().equals(groupName)).count();
        if (groupcount == 0) {
            phoenixAlarmStream.createGroup(groupName);
        }

        Thread thread = new Thread(() -> {
            while (isRunning) {
                try {
                    // 获取并移除此队列的头部，在元素变得可用之前一直等待 。queue的长度 == 0 的时候，阻塞等待
                    Map<StreamMessageId, Map<String, String>> messageIdMapMap = phoenixAlarmStream.readGroup(groupName, consumerName,
                            StreamReadGroupArgs.neverDelivered().timeout(Duration.of(30, ChronoUnit.SECONDS)).count(20));
//                    log.info("phoenixAlarmStream consume message = {}", messageIdMapMap);
                    if (messageIdMapMap == null) {
                        continue;
                    }
                    for (StreamMessageId messageId : messageIdMapMap.keySet()) {
                        log.info("messageId:{}", messageId);
                        Map<String, String> messageMap = messageIdMapMap.get(messageId);
                        log.info("message:{}", messageIdMapMap.get(messageId));

                        emsThreadPoolTaskExecutor.execute(new Runnable() {
                            @Override
                            public void run() {
                                phoenixAlarmService.onMessage(messageMap);
                            }
                        });

                        phoenixAlarmStream.ack(groupName, messageId);
                        phoenixAlarmStream.remove(messageId);
                    }
                } catch (Exception e) {
                    log.error("phoenixAlarmStream异常", e);
                    ThreadUtil.sleep(1000);
                }
            }
        });

        thread.setDaemon(true);
        thread.start();

    }


}
